package dse.messages.singleversion;

import java.util.Collections;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.TreeMap;
import java.util.TreeSet;

import dse.DataObject;
import dse.Experiment;
import dse.ExperimentRecorder;
import dse.Node;
import dse.Version;
import dse.messages.Message;
import dse.messages.MessagePurpose;
import dse.messages.MessageUtil;
import dse.messages.Transaction;
import dse.messages.multiversion.VersionReader;

public class GetNSRTransaction extends Transaction implements VersionReader,
		ShareDownloader {
	private DataObject desiredObject;
	private LinkedList<Node> residentNodes;
	private boolean failed;
	private TreeMap<Version, LinkedList<Node>> sharesDownloaded;
	private LinkedList<Version> candidateVersion;
	private int requestShareCount;
	private int validShareCount;
	private int maxFailure;
	private int successCount;
	private int failureCount;
	//private TreeSet<Integer> ts;
	private Version maxVersion;
	private int sharesToBeUpdate;
	private boolean hasRequestPreviousShares;
	private int maxTS;
	private HashSet<Node> hasSent;
	
	public GetNSRTransaction(Experiment experiment,
			int id,
			Node requester,
			DataObject desiredObject,
			double time
			)
	{
		super(id, requester, experiment, time);
		
		this.desiredObject = desiredObject;
		this.failed = false;
		this.requestShareCount = 0;
		this.validShareCount = 0;
		this.maxFailure = experiment.modules().nodeEventScheduler().getEventsPerPeriod();
		this.successCount = 0;
		this.failureCount = 0;
		this.maxVersion = null;
		this.sharesToBeUpdate = desiredObject.shareCount();
		/*if(isUpdate && isPartialUpdate)
		//{
			this.sharesToBeUpdate = (int)(desiredObject.shareCount()*Math.random());
			if(this.sharesToBeUpdate == 0)
				this.sharesToBeUpdate = 1;			
		}*/
		this.hasRequestPreviousShares = false;
		this.maxTS = 0;
	}

	@Override
	public void start(ExperimentRecorder recorder) {
		// TODO Auto-generated method stub
		residentNodes = MessageUtil.getNodesByDht(experiment.network(), requester, desiredObject, experiment.modules().assigner());		

		if(isUpdate)
		{
			recorder.dataUpdateInitiated(super.id, requester, desiredObject, startTime);
			if(ExperimentRecorder.debug)
				recorder.debug(this, "start", startTime);
			this.requestShareCount = desiredObject.shareCount();
			this.validShareCount = requestShareCount - this.maxFailure;
			
			for (int i=0; (i<requestShareCount) && (i<residentNodes.size()); i++)
				sendMaxTSRequest(recorder, residentNodes.get(i), startTime);
		}
		else
		{
			recorder.dataRequestInitiated(super.id, requester, desiredObject, startTime);
			recorder.debug(this, "start", startTime);
			
			//this.requestShareCount = this.desiredObject.requiredShareCount() + this.maxFailure;
			this.requestShareCount = experiment.modules().shareRequestDecisionMaker().sharesToRequestFor(desiredObject);
			this.validShareCount = desiredObject.requiredShareCount();
			validateRequestShareCount(residentNodes.size());
			
			this.hasSent = new HashSet<Node>();
			sharesDownloaded = new TreeMap<Version, LinkedList<Node>>();
			
			for(int i=0; i<requestShareCount; i++)
			{
				sendGetLCShareRequest(recorder, residentNodes.get(i), startTime);
			}		
		}
	}
	
	private void validateRequestShareCount(int iv){
		if(requestShareCount > iv)
			requestShareCount = iv;
	}
	
	private void sendMaxTSRequest(ExperimentRecorder recorder, Node node, Double time)
	{
		if (node == null) {
			// We should turn this into a warning ...
			recorder.warn(
					"Share assigner failed to map share to node in cluster!"
					);
			return;
		}
		
		Message m = super.createRequestMaxTSMessage(requester, node, desiredObject, 
				MessagePurpose.DataObjectUpdate, this);
		if(ExperimentRecorder.debug)
			recorder.debug(this, "new msg: " + m, time);
		experiment.messageQueue().enqueueMessage(m, time);
	}
	
	private void sendShareWritingRequest(ExperimentRecorder recorder, Node node, Version v, Double time)
	{
		if (node == null) {
			// We should turn this into a warning ...
			recorder.warn("Share assigner failed to map share to node in cluster!");
			return;
		}
		
		Message m = super.createShareUpdateOSVMessage(requester, 
				node, desiredObject, v, MessagePurpose.DataObjectUpdate, this);
		if(ExperimentRecorder.debug)
			recorder.debug(this, "new msg: " + m, time);
		experiment.messageQueue().enqueueMessage(m, time);
	}
	
	private void sendGetLCShareRequest(ExperimentRecorder recorder, Node node, double time){
		if (node == null) {
			// We should turn this into a warning ...
			recorder.warn(
					"Share assigner failed to map share to node in cluster!"
					);
			return;
		}
		
		Message m = super.createRequestLCShareMessage(requester, node, desiredObject, 
				MessagePurpose.DataObjectRequest, this);
		if(ExperimentRecorder.debug)
			recorder.debug(this, "new msg: " + m, time);
		experiment.messageQueue().enqueueMessage(m, time);
	}

	@Override
	public Transaction getTransaction(){
		return this;
	}

	@Override
	public void readVersion(ExperimentRecorder recorder,TreeSet<Version> versions, double time, Node from) {}

	@Override
	public void readVersion(ExperimentRecorder recorder, Node from,	Version version, double time) {
		// TODO Auto-generated method stub
		if(failed || finished)
			return;
		
		if(version == null){
			if(!this.hasSent.contains(from)){
				failureCount++;
				this.hasSent.add(from);
			}
			return;
		}
		
		if(!this.hasSent.contains(from)){
			this.hasSent.add(from);
			successCount++;
		}
		if(this.sharesDownloaded.get(version) == null)
			this.sharesDownloaded.put(version, new LinkedList<Node>());
		if(!this.sharesDownloaded.get(version).contains(from))
			this.sharesDownloaded.get(version).add(from);
		
		if(this.hasRequestPreviousShares && (this.sharesDownloaded.get(version).size() >= validShareCount)){
			transactionSucceed(recorder, time);
			return;
		}
		
		if(candidateVersion == null)
			candidateVersion = new LinkedList<Version>();
		candidateVersion.addLast(version);
		
		if(successCount + failureCount >= requestShareCount){
			if(successCount < validShareCount){
				int begin = requestShareCount;
				requestShareCount = validShareCount + maxFailure;
				validateRequestShareCount(residentNodes.size());
				for(int i=begin; i<requestShareCount; i++)
				{
					sendGetLCShareRequest(recorder, residentNodes.get(i), time);
				}
			}
			else{
				processVersion(recorder, time);
			}			
		}
	}
	
	private void processVersion(ExperimentRecorder recorder, double timeLeft)
	{
		if((sharesDownloaded==null) || (sharesDownloaded.size()==0))
		{
			transactionFailed(recorder, timeLeft);
			return;
		}
		
		maxVersion = sharesDownloaded.lastKey();
		if(maxVersion.getSequence()>requester.getSequence())
			requester.setSequence(maxVersion.getSequence());
		
		/*Map.Entry<Version, LinkedList<Node>> vn = sharesDownloaded.pollLastEntry();
		while((vn.getValue().size()<validShareCount) && (sharesDownloaded.size()>0))
		{
			vn = sharesDownloaded.pollLastEntry();
		}*/
		for(Version vn : sharesDownloaded.descendingKeySet()){
			if(sharesDownloaded.get(vn).size() >= validShareCount)//== requestShareCount)
			{
				transactionSucceed(recorder, timeLeft);
				return;
			}
			/*else if(vn.getValue().size() < requestShareCount && vn.getValue().size()>=validShareCount)
			{
				maxVersion = vn.getKey();
				for(Version version : sharesDownloaded.keySet())
				{
					for(Node node : sharesDownloaded.get(version))
					{
						sendUpdateShareRequest(recorder, node, maxVersion, timeLeft);
					}
				}
				transactionSucceed(recorder, timeLeft);	
			}*/
		}
		
		if(!this.hasRequestPreviousShares)
			//recorder.requestPreviousShares(desiredObject, false);
		//else{
			this.hasRequestPreviousShares = true;
			//recorder.requestPreviousShares(desiredObject, true);
		//}
		
		failureCount = 0;
		successCount = 0;
		
		//this.round++;
		//this.sharesDownloaded.clear();
		recorder.OneRoundInMoreFinished(super.id, timeLeft);
		//System.out.println("Client:" + requester.id());
		processVersions(recorder, timeLeft);
	}
	
	private void processVersions(ExperimentRecorder recorder, double timeLeft)
	{
		if((candidateVersion==null) || (candidateVersion.size()==0))
		{
			transactionFailed(recorder, timeLeft);
			return;
		}
		Collections.sort(candidateVersion);
		
		//while(candidateVersion.size() > 1){
			candidateVersion.pollFirst();
		//}
		
		maxVersion = candidateVersion.pollFirst();
		if(requestShareCount < validShareCount + maxFailure)
			requestShareCount = validShareCount + maxFailure;
		
		validateRequestShareCount(residentNodes.size());
		for(int i=0; i<this.requestShareCount; i++)
			sendRequestShareWithVersionRequest(recorder, maxVersion, residentNodes.get(i), timeLeft);
	}
	
	private void sendRequestShareWithVersionRequest(ExperimentRecorder recorder, Version version, Node node, double time){
		if (node == null) {
			// We should turn this into a warning ...
			recorder.warn(
					"Share assigner failed to map share to node in cluster!"
					);
			return;
		}
		
		Message m = super.createRequestShareWithVersionMessage(requester, node, desiredObject, version,
				MessagePurpose.DataObjectRequest, this, true);
		if(ExperimentRecorder.debug)
			recorder.debug(this, "new msg: " + m, time);
		experiment.messageQueue().enqueueMessage(m, time);
	}

	@Override
	public void readVersion(ExperimentRecorder recorder, int seq, double time) {
		// TODO Auto-generated method stub
		if(seq == -1)
		{
			failureCount++;
			return;
		}
		successCount++;
		if(this.maxTS < seq)
			this.maxTS = seq;
		if(successCount+failureCount >= requestShareCount){
			successCount = 0;
			failureCount = 0;
			if(this.maxTS >= requester.getSequence())
				requester.setSequence(this.maxTS);
			Version version = new Version(requester.sequence(), requester.id());
			for(int i=0; (i<this.sharesToBeUpdate) && (i<residentNodes.size()); i++)
				sendShareWritingRequest(recorder, residentNodes.get(i), version, time);
		}
	}

	@Override
	public void readVersion(ExperimentRecorder recorder, int seq, int index, double time) {	}
	
	/*private void sendVersion(ExperimentRecorder recorder, Node node, Version v, double time){
		Message m = super.createSendVersionMessage(requester, 
				node, desiredObject, maxVersion, MessagePurpose.DataObjectUpdate, this);
		if(ExperimentRecorder.debug)
			recorder.debug(this, "new msg: " + m, time);
		experiment.messageQueue().enqueueMessage(m, time);
	}
	
	private void sendMAC(ExperimentRecorder recorder, Node node, Version v, double time){
		Message m = super.createSendMACMessage(requester, 
				node, desiredObject, maxVersion, MessagePurpose.DataObjectUpdate, this);
		if(ExperimentRecorder.debug)
			recorder.debug(this, "new msg: " + m, time);
		experiment.messageQueue().enqueueMessage(m, time);
	}*/

	@Override
	public void readVersion(ExperimentRecorder recorder, double time,
			LinkedList<Version> versions, Version version, Node from) {
		// TODO Auto-generated method stub

	}

	@Override
	public boolean isDoneDownloadingShares() {
		// TODO Auto-generated method stub
		return (failed || finished);
	}

	@Override
	public boolean isDoneUpdatingShares() {
		// TODO Auto-generated method stub
		return (failed || finished);
	}

	@Override
	public void shareDownloadFailed(ExperimentRecorder recorder,
			DataObject dobj, int shareIndex, double timeLeft) {
		// TODO Auto-generated method stub

	}

	@Override
	public void shareDownloaded(ExperimentRecorder recorder, Node from,
			DataObject dobj, int shareIndex, double timeLeft) {
		// TODO Auto-generated method stub

	}

	@Override
	public void shareUpdateFailed(ExperimentRecorder recorder, DataObject dobj,
			int shareIndex, double timeLeft) {
		// TODO Auto-generated method stub
		if(isUpdate) {
			this.failureCount++;
			if((successCount + failureCount) >= this.sharesToBeUpdate)
			{
				if(this.successCount >= validShareCount)
					transactionSucceed(recorder, timeLeft);
				else
					transactionFailed(recorder, timeLeft);
			}
		}
	}

	@Override
	public void shareUpdated(ExperimentRecorder recorder, Node to,
			DataObject dobj, int shareIndex, double timeLeft) {
		// TODO Auto-generated method stub
		if(isUpdate) {		
			this.successCount++;
			if((successCount+failureCount) >= this.sharesToBeUpdate)
			{
				if(this.successCount >= validShareCount)
					transactionSucceed(recorder, timeLeft);
				else
					transactionFailed(recorder, timeLeft);
			}
		}
	}
	
	private void transactionSucceed(ExperimentRecorder recorder, double timeLeft)
	{
		finished = true;
		failed  = false;
		//System.out.println("Data ID : " + desiredObject.id() + "Version : " + maxVersion.getSequence() + " , " + maxVersion.getClientID());
		if(isUpdate)
		{
			recorder.debug(
					"GetOSVTransaction trans#" + super.id + 
					" is finshed!!!"
					);
			recorder.dataUpdateFinished(super.id, requester, desiredObject, true, timeLeft);
		}
		else
		{
			for(int i=0; i<requestShareCount; i++)
			{
				Message m = super.createReadCompleteMessage(requester, this.residentNodes.get(i), desiredObject, MessagePurpose.DataObjectRequest, this, false);
				recorder.debug(this, "new msg: " + m, timeLeft);
				experiment.messageQueue().enqueueMessage(m, timeLeft);
			}
			recorder.debug(
					"GetOSVTransaction trans#" + super.id + 
					" is finshed!!!"
					);
			recorder.dataRequestFinished(super.id, requester, desiredObject, true, this.residentNodes.get(this.validShareCount-1),
					this.hasRequestPreviousShares, timeLeft);
		}
	}
	
	private void transactionFailed (ExperimentRecorder recorder, double timeLeft)
	{
		recorder.debug(this, "transFailed", timeLeft);
		
		finished = true;
		failed = true;
		if(isUpdate)
			recorder.dataUpdateFinished(super.id, requester, desiredObject, false, timeLeft);
		else{
			for(int i=0; i<requestShareCount; i++)
			{
				Message m = super.createReadCompleteMessage(requester, this.residentNodes.get(i), desiredObject, MessagePurpose.DataObjectRequest, this, false);
				recorder.debug(this, "new msg: " + m, timeLeft);
				experiment.messageQueue().enqueueMessage(m, timeLeft);
			}
			recorder.dataRequestFinished(super.id, requester, desiredObject, false, null, this.hasRequestPreviousShares, timeLeft);
		}
	}
}
